package kafka.loganalysis;

import kafka.KafkaProperties;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @author: pizhihui
 * @datae: 2017-11-09
 */
public class LogConsumer {

    private static final String GROUP_ID = "log_group2";
    private static final String CONSUMER_ID = "consumer1";
    private static String groupId;
    private static String consumerId;
    private int threadCount;


    public LogConsumer(String grouId, String consumerId, int threadCount) {
        this.groupId = grouId;
        this.consumerId = consumerId;
        this.threadCount = threadCount;
    }

    public void consume() {
        Properties props = new Properties();
        props.put("group.id", groupId);
        props.put("consumer.id", consumerId);
        props.put("zookeeper.connect", KafkaProperties.ZK_CONNECT);
        props.put("zookeeper.session.timeout.ms", "60000");
        props.put("zookeeper.sync.time.ms", "2000");

        ConsumerConfig config = new ConsumerConfig(props);
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);

        Map<String, Integer> topicCountMap = new HashMap<>();
        // 设置topic的线程
        topicCountMap.put(LogGenerator.LOG_TOPIC, threadCount);
        Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumer.createMessageStreams(topicCountMap);

        List<KafkaStream<byte[], byte[]>> kafkaStreams = streams.get(LogGenerator.LOG_TOPIC);

        for (KafkaStream<byte[], byte[]> stream : kafkaStreams) {
            new KafkaStreamThread(stream).start();
        }
    }



    private static class KafkaStreamThread extends Thread {

        KafkaStream<byte[], byte[]> stream;

        public KafkaStreamThread(KafkaStream<byte[], byte[]> stream) {
            this.stream = stream;
        }

        @Override
        public void run() {
            ConsumerIterator iterator = stream.iterator();
            while (iterator.hasNext()) {
                MessageAndMetadata message = iterator.next();
                String topic = message.topic();
                int partition = message.partition();
                long offset = message.offset();
                //String key = new String(message.key());
                String msg = new String(message.message().toString());
                // 在这里处理消息,这里仅简单的输出
                // 如果消息消费失败，可以将已上信息打印到日志中，活着发送到报警短信和邮件中，以便后续处理
                System.out.println("consumerid:" + consumerId + ", thread : " + Thread.currentThread().getName()
                        + ", topic : " + topic + ", partition : " + partition + ", offset : " + offset + " , key : "
                        + message.key() + " , mess : " + msg);
            }
        }
    }




    public static void main(String[] args) {
        String groupid = "log_consumer_group";
        LogConsumer consumer1 = new LogConsumer(groupid, "myconsumer1", 1);
        LogConsumer consumer2 = new LogConsumer(groupid, "myconsumer2", 1);

        consumer1.consume();
        consumer2.consume();


    }

}
